package com.wyh.apitest.transformation.multithread

import com.wyh.apitest.source.SensorReading
import org.apache.flink.streaming.api.scala._

/**
  * 1．Union 之前两个流的类型必须是一样，Connect 可以不一样，在之后的 coMap 中再去调整成为一样的。
  * 2. Connect 只能操作两个流，Union 可以操作多个。
  */
object UnionDemo {
  def main(args: Array[String]): Unit = {
    //定义流式处理环境
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)

    val sensorinfo: DataStream[String] = streamEnv.readTextFile("D:\\IdeaProjects\\FlinkStudyScala\\src\\main\\resources\\sensor.txt")
    //转换成样例类类型
    val dataStream = sensorinfo.map(s => {
      val arr: Array[String] = s.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    //多流转换算子----分流操作(在ProessFunction的代码里讲解另一种解法侧输出流)
    val splitStream: SplitStream[SensorReading] = dataStream.split(data => {
      if (data.temperature > 30.0) {
        Seq("higth")
      } else {
        Seq("low")
      }
    })
    //挑选出温度大于30的
    val higthTempStream: DataStream[SensorReading] = splitStream.select("higth")
    val warnTempStream: DataStream[(String, Double)] = higthTempStream.map(data => {
      (data.id, data.temperature)
    })
    //挑选出温度小于30的
    val lowTempStream: DataStream[SensorReading] = splitStream.select("low")


    //多流转换算子----Union合流操作（数据类型要一致,但是可以传多条相同数据类型的流）
//    warnTempStream.union(lowTempStream) //报错，因为数据类型不一致
    val unionTempStream: DataStream[SensorReading] = higthTempStream.union(lowTempStream)

    unionTempStream.print("Union")


    streamEnv.execute("多流转换算子之Union")

  }

}